package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/gogo/protobuf/proto"
	"github.com/tendermint/tendermint/my_app/config"
	ql "github.com/tendermint/tendermint/my_app/querylayer"
	"github.com/tendermint/tendermint/my_app/utils"
	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
	tmhttp "github.com/tendermint/tendermint/rpc/client/http"
	"github.com/tendermint/tendermint/types"
)

type Server struct {
	engine         *gin.Engine
	app            *ql.QueryEngine
	cli            *tmhttp.HTTP
	curHeight      int64
	blocktoProcess []*types.Block
	blockTxs       []*tmproto.Tx
	mu             sync.Mutex
	processingTxs  bool

	//===================================================
	AvailableHeight   int64
	Lastprocessheight int64   //上一个处理的区块高度
	db                *sql.DB //存储交易数据

}

type BlockBatch struct {
	Heights       map[int64]bool // 当前已收集完整的的高度列表
	Timer         *time.Timer    // 超时计时器
	CurrentHeight int64          // 当前正在处理的高度
	BatchSize     int            // 每次处理的区块数量
	Timeout       time.Duration  // 超时时间
}

func NewServer() *Server {

	engine := gin.Default()
	//demo
	cli, err := tmhttp.New("http://127.0.0.1:26657", "/websocket")
	if err != nil {
		panic(err)
	}
	server := &Server{
		cli:       cli,
		curHeight: 2,
		engine:    engine,
		app:       ql.NewQueryEngine(),

		Lastprocessheight: 0,
		AvailableHeight:   0,
	}
	server.db = server.initCurHeight()
	server.registerRoutes()

	return server
}

func (s *Server) initCurHeight() *sql.DB {
	// 数据库连接字符串
	dsn := "user=demo_user password=123456 host=localhost port=5432 dbname=demodb sslmode=disable"
	db, err := sql.Open("postgres", dsn)
	if err != nil {
		log.Fatal(err)
	}

	// 检查表是否存在，如果不存在则创建表
	tableCheckQuery := `SELECT EXISTS (
		SELECT 1 FROM information_schema.tables 
		WHERE table_schema = 'public' AND table_name = 'current_height'
	);`
	var exists bool
	err = db.QueryRow(tableCheckQuery).Scan(&exists)
	if err != nil {
		log.Fatal(err)
	}

	if !exists {
		// 如果表不存在，创建表并初始化当前高度为2
		createTableQuery := `CREATE TABLE current_height (
			height INT NOT NULL
		);`
		_, err = db.Exec(createTableQuery)
		if err != nil {
			log.Fatal(err)
		}
		// 初始化当前高度为2
		insertInitialHeight := `INSERT INTO current_height (height) VALUES ($1)`
		_, err = db.Exec(insertInitialHeight, 2)
		if err != nil {
			log.Fatal(err)
		}
		s.curHeight = 2
		fmt.Println("Table created and current height initialized to 2.")
	} else {
		// 如果表存在，获取当前高度
		var currentHeight int64
		getHeightQuery := `SELECT height FROM current_height limit 1;`
		err = db.QueryRow(getHeightQuery).Scan(&currentHeight)
		if err != nil {
			log.Fatal(err)
		}
		s.curHeight = currentHeight
		fmt.Printf("Current height: %d\n", currentHeight)
	}
	return db
}

func (s *Server) Updatecurheight(newHeight int64) {
	// 假设你想更新当前高度为当前高度 + 1
	updateHeightQuery := `UPDATE current_height SET height = $1;`
	_, err := s.db.Exec(updateHeightQuery, newHeight)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Current height updated to: %d\n", newHeight)
}
func (s *Server) statusHandler(c *gin.Context) {
	if s.processingTxs {
		c.JSON(200, gin.H{
			"message": "Server is processing transactions, try again later.",
		})
		return
	}
	c.JSON(200, gin.H{
		"message": "Server is ready.",
	})
}

func (s *Server) buildIndex(c *gin.Context) {
	if s.processingTxs {
		c.JSON(200, gin.H{
			"message": "Server is processing transactions, try again later.", "code": 1,
		})
		return
	}

	var params struct {
		DataType string `json:"data_type" binding:"required"`
		CateName string `json:"cate_name" binding:"required"`
	}
	if err := c.ShouldBindJSON(&params); err != nil {
		c.JSON(200, gin.H{"message": err.Error(), "code": 1})
		return
	}

	s.app.BuildIndex(params.DataType, params.CateName)
	c.JSON(200, gin.H{"code": 0})
}

func (s *Server) showIndex(c *gin.Context) {
	indices := s.app.ShowIndex()
	c.JSON(200, gin.H{"indices": indices})
}

func (s *Server) sqlQuery(c *gin.Context) {
	var params struct {
		Sql string `json:"sql" binding:"required"`
	}
	if err := c.ShouldBindJSON(&params); err != nil {
		c.JSON(200, gin.H{"message": err.Error(), "code": 1})
		return
	}

	results := s.app.Sql_Query(params.Sql)
	fmt.Println("===================", params.Sql)
	fmt.Println(">>>>>>>>>>>>>>>>>>>>", results)
	c.JSON(http.StatusOK, gin.H{"data": results, "code": 0})

}

func (s *Server) PictureQueryVectors(c *gin.Context) {
	log.Println("process knnquery=============")
	if s.processingTxs {
		c.JSON(200, gin.H{
			"message": "Server is processing transactions, try again later.", "code": 1,
		})
		return
	}

	//input:[]byte
	file, err := c.FormFile("image")
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to retrieve file: " + err.Error()})
		return
	}

	// 打开文件以读取内容
	fileContent, err := file.Open()
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to open file: " + err.Error()})
		return
	}
	defer fileContent.Close()

	//特征提取
	image, err := utils.ConvertToFile(fileContent)
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed Convert To File: " + err.Error(), "code": 1})
		return
	}
	vector, err := utils.Colorhistfeature(image)
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to extract feature: " + err.Error(), "code": 1})
		return
	}

	_, vectors, _ := s.app.KnnQuery("image", "arm", vector, 20, 0)

	c.JSON(http.StatusOK, gin.H{"data": vectors, "code": 0})

}

func (s *Server) KnnQueryVectors(c *gin.Context) {
	log.Println("process knnquery=============")
	if s.processingTxs {
		c.JSON(200, gin.H{
			"message": "Server is processing transactions, try again later.", "code": 1,
		})
		return
	}

	rawData, err := c.GetRawData()
	if err != nil {
		fmt.Println("GetRawData Error==================")
		c.JSON(200, gin.H{"message": err.Error(), "code": 1})
		return
	}

	fmt.Println(string(rawData))

	var params struct {
		Eigenvalue   []float32 `json:"eigenvalue"`
		FileType     string    `json:"filetype"`
		FileCategory string    `json:"filecategory"`
		K            int32     `json:"k"`
		Ef           int32     `json:"ef"`
	}

	if err := json.Unmarshal(rawData, &params); err != nil {
		c.JSON(200, gin.H{
			"code":    1,
			"message": err.Error(),
		})
		return
	}

	_, vectors, _ := s.app.KnnQuery(params.FileType, params.FileCategory, params.Eigenvalue, params.K, 0)

	c.JSON(http.StatusOK, gin.H{"results": vectors, "code": 0})

}

func (s *Server) knnQuery(c *gin.Context) {
	log.Println("process knnquery=============")
	if s.processingTxs {
		c.JSON(200, gin.H{
			"message": "Server is processing transactions, try again later.", "code": 1,
		})
		return
	}

	var params struct {
		Eigenvalue   []float32 `json:"eigenvalue" binding:"required"`
		FileType     string    `json:"filetype" binding:"required"`
		FileCategory string    `json:"filecategory" binding:"required"`
		K            int32     `json:"k" binding:"required"`
		Ef           int32     `json:"ef" binding:"required"`
	}
	if err := c.ShouldBindJSON(&params); err != nil {
		c.JSON(200, gin.H{"message": err.Error(), "code": 1})
		return
	}

	results, _, _ := s.app.KnnQuery(params.FileType, params.FileCategory, params.Eigenvalue, params.K, 0)

	c.JSON(200, gin.H{"data": results, "code": 0})
}

func main() {

	server := NewServer()
	// 创建或打开 log 文件
	file, err := os.OpenFile("my_app/mylog.log", os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
	if err != nil {
		log.Fatal(err)
	}
	// 设置日志输出到文件
	log.SetOutput(file)
	defer file.Close()

	// 输出日志
	log.Println("This is a log message.")

	server.Run("0.0.0.0:9090")

}

func (s *Server) Run(addr string) {
	// demo
	go s.startBlockSync(config.SyncTimeInterval * time.Second)
	s.engine.Run(addr)
}

func (s *Server) registerRoutes() {
	s.engine.POST("/status", s.statusHandler)
	s.engine.POST("/build_index", s.buildIndex)
	s.engine.POST("/show_index", s.showIndex)
	s.engine.POST("/knn_query", s.knnQuery)
	s.engine.POST("/sql_query", s.sqlQuery)
	s.engine.POST("/demo", s.demo)
	s.engine.POST("/pic_queryvec", s.PictureQueryVectors)
	s.engine.POST("/knn_queryvec", s.KnnQueryVectors)
}

func (s *Server) demo(c *gin.Context) {
	//input:[]byte
	file, err := c.FormFile("image")
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to retrieve file: " + err.Error()})
		return
	}

	// 打开文件以读取内容
	fileContent, err := file.Open()
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to open file: " + err.Error()})
		return
	}
	defer fileContent.Close()

	// 将文件内容读取为字节数组
	// fileBytes, err := io.ReadAll(fileContent)

	// if err != nil {
	// 	c.JSON(200, gin.H{"error": "Failed to read file: " + err.Error(), "code": 1})
	// 	return
	// }
	//特征提取
	image, err := utils.ConvertToFile(fileContent)
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed Convert To File: " + err.Error(), "code": 1})
		return
	}
	vector, err := utils.Colorhistfeature(image)
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to extract feature: " + err.Error(), "code": 1})
		return
	}
	fmt.Println(vector)
	//knn查询
	results, _, _ := s.app.KnnQuery("image", "arm", vector, 80, 30)
	//sql查询
	// 原始 SQL 语句
	sqlQuery1 := "SELECT Province, COUNT(1) AS Person_Count FROM User_table JOIN file_data ON userid = owner_id WHERE object_id IN ('101', '102', '103', '104', '105') GROUP BY Province ORDER BY Province;"
	sqlQuery2 := "SELECT age, COUNT(1) AS Person_Count FROM User_table JOIN file_data ON userid = owner_id WHERE object_id IN ('101', '102', '103', '104', '105') GROUP BY age ORDER BY age;"

	// 替换 SQL 中的占位符
	Query1 := utils.ReplaceSQLPlaceholders(sqlQuery1, results) //
	Query2 := utils.ReplaceSQLPlaceholders(sqlQuery2, results)
	fmt.Println(Query1)
	fmt.Println(Query2)
	//返回两个分布

	pd := s.app.Sql_Query(Query1)
	ad := s.app.Sql_Query(Query2)

	djson, err := utils.IntegrateDistributions(pd, ad)
	if err != nil {
		c.JSON(200, gin.H{"message": "Failed to Integrate Distributions: " + err.Error(), "code": 1})
		return
	}

	c.JSON(http.StatusOK, gin.H{"data": djson, "code": 0})
}

// demo
func (s *Server) startBlockSync(interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for range ticker.C {

		abciQueryRes, err := s.cli.ABCIQuery(context.TODO(), "", nil)
		if err != nil {
			fmt.Println("Error querying ABCI:", err)
			continue
		}

		height := abciQueryRes.Response.Height
		fmt.Printf("Current height: %d, Latest height: %d\n", s.curHeight, height)
		//判断是否超过size
		if height > s.curHeight+config.BatchSize {
			for h := s.curHeight + 1; h <= height; h++ {
				blockRes, err := s.cli.Block(context.TODO(), &h)
				if err != nil {
					fmt.Println("Error fetching block:", err)
					continue
				}
				fmt.Printf("========= fetching block height %d =========\n", h)
				s.mu.Lock()
				s.blocktoProcess = append(s.blocktoProcess, blockRes.Block)
				s.mu.Unlock()
			}

			// for _, txpb := range blockRes.Block.Data.Txs {
			// 	tx := new(tmproto.Tx)
			// 	if err := proto.Unmarshal(txpb, tx); err != nil {
			// 		fmt.Println("Error unmarshaling tx:", err)
			// 		continue
			// 	}

			// 	// 存储交易信息
			// 	s.mu.Lock()
			// 	s.blockTxs = append(s.blockTxs, tx)
			// 	s.mu.Unlock()
			// }

			// 如果积累了足够多的交易，处理它们
			// if len(s.blockTxs) >= config.BatchTxSize {
			// 	s.processTransactions()
			// 	// 更新当前高度
			// 	s.curHeight = h - 1
			// }

			if len(s.blocktoProcess) >= config.BatchSize {
				s.curHeight = height
				s.Updatecurheight(s.curHeight)
				s.processTransactions()
				// 更新当前高度

			}
		}

	}
}

func (s *Server) processTransactions() {
	s.mu.Lock()
	// if len(s.blockTxs) < config.BatchTxSize {
	// 	// 如果交易不足 xx 个，直接返回
	// 	s.mu.Unlock()
	// 	return
	// }

	if len(s.blocktoProcess) < config.BatchSize {
		// 如果交易不足 xx 个，直接返回
		s.mu.Unlock()
		return
	}
	s.processingTxs = true
	// 复制要处理的交易并清空原有队列
	bToProcess := s.blocktoProcess[:len(s.blocktoProcess)-1]
	s.blocktoProcess = s.blocktoProcess[len(s.blocktoProcess)-1:]

	// txsToProcess := s.blockTxs[:config.BatchTxSize]
	// s.blockTxs = s.blockTxs[config.BatchTxSize:]

	s.mu.Unlock()

	fmt.Println("Processing transactions...")
	for i, block := range bToProcess {
		fmt.Printf("Processing block no %d...,height=%d\n", i, block.Height)
		for _, txpb := range block.Data.Txs {
			tx := new(tmproto.Tx)
			if err := proto.Unmarshal(txpb, tx); err != nil {
				fmt.Println("Error unmarshaling tx:", err)
				continue
			}
			if err := s.app.DeliverTx(tx); err != nil {
				fmt.Printf("Error delivering tx: %+v\n", err)
			}
		}

	}
	fmt.Println("Finished processing transactions.")

	// 标记为可用状态
	s.mu.Lock()
	s.processingTxs = false
	s.mu.Unlock()

	// 再次检查是否还有 xxx 笔交易需要处理
	s.processTransactions()
}
